package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TrafficHomeDetailPageViewBea;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.xml.crypto.Data;
import java.time.Duration;
import java.util.Iterator;

/**
 * @author shenyucheng
 * @version 1.0
 * 12.3流量域首页、详情页页面浏览各窗口汇总表
 * @description: TODO
 * @date 2025/3/3 10:06
 */
public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficHomeDetailPageViewWindow().start(10023,1,Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW ,Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //1.ETL数据
        SingleOutputStreamOperator<JSONObject> etlStream = getEtlStream(stream);
        //2、根据MID分组，
        KeyedStream<JSONObject, String> keyedMid = etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });
        //计算浏览量
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBea> process =  extracted(keyedMid);
        //3、添加水位求HomeUV和DetailUV
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBea> reduceStream = getReduce(process);
        reduceStream.print();
        reduceStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW));


    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBea> getReduce(SingleOutputStreamOperator<TrafficHomeDetailPageViewBea> process) {
        return process.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficHomeDetailPageViewBea>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficHomeDetailPageViewBea>() {
                    @Override
                    public long extractTimestamp(TrafficHomeDetailPageViewBea trafficHomeDetailPageViewBea, long l) {
                        return trafficHomeDetailPageViewBea.getTs();
                    }
                })).windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5))).reduce(new ReduceFunction<TrafficHomeDetailPageViewBea>() {
            @Override
            public TrafficHomeDetailPageViewBea reduce(TrafficHomeDetailPageViewBea t1, TrafficHomeDetailPageViewBea t2) throws Exception {
                t1.setHomeUvCt(t1.getHomeUvCt() + t2.getHomeUvCt());
                t1.setGoodDetailUvCt(t1.getGoodDetailUvCt() + t2.getGoodDetailUvCt());
                return t1;
            }
        }, new ProcessAllWindowFunction<TrafficHomeDetailPageViewBea, TrafficHomeDetailPageViewBea, TimeWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<TrafficHomeDetailPageViewBea, TrafficHomeDetailPageViewBea, TimeWindow>.Context context, Iterable<TrafficHomeDetailPageViewBea> iterable, Collector<TrafficHomeDetailPageViewBea> collector) throws Exception {
                TimeWindow window = context.window();
                String start = DateFormatUtil.tsToDate(window.getStart());
                String end = DateFormatUtil.tsToDate(window.getEnd());
                String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                Iterator<TrafficHomeDetailPageViewBea> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    TrafficHomeDetailPageViewBea trafficHomeDetailPageViewBea = iterator.next();
                    trafficHomeDetailPageViewBea.setStt(start);
                    trafficHomeDetailPageViewBea.setEdt(end);
                    trafficHomeDetailPageViewBea.setCurDate(curDate);
                    collector.collect(trafficHomeDetailPageViewBea);
                }
            }
        });
    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBea> extracted(KeyedStream<JSONObject, String> keyedMid) {
        return keyedMid.process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBea>() {
            private ValueState<String> homeState;
            private ValueState<String> detailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> homeStateDesc = new ValueStateDescriptor<>("home_state", String.class);
                ValueStateDescriptor<String> detailStateDesc = new ValueStateDescriptor<>("detail_state", String.class);
                homeStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                detailStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                homeState = getRuntimeContext().getState(homeStateDesc);
                detailState = getRuntimeContext().getState(detailStateDesc);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBea>.Context context, Collector<TrafficHomeDetailPageViewBea> collector) throws Exception {
                // 处理逻辑
                String homeValue = homeState.value();
                String detailValue = detailState.value();
                String curDt = DateFormatUtil.tsToDate(jsonObject.getLong("ts"));
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                long homeCt = 0L;
                long detailCt = 0L;
                //页面浏览量
                if ("home".equals(pageId)) {
                    if (!curDt.equals(homeValue)) {
                        homeCt = 1L;
                        homeState.update(curDt);
                    }
                } else {
                    //详情页页面浏览量
                    if (!curDt.equals(detailValue)) {
                        detailCt = 1L;
                        //修改状态的值
                        detailState.update(curDt);
                    }
                }
                if (homeCt != 0 || detailCt != 0) {
                    collector.collect(TrafficHomeDetailPageViewBea.builder()
                            .homeUvCt(homeCt)
                            .goodDetailUvCt(detailCt)
                            .ts(jsonObject.getLong("ts"))
                            .build());
                }

            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getEtlStream(DataStreamSource<String> stream) {
        return stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String mid = jsonObject.getJSONObject("common").getString("mid");
                String ts = jsonObject.getString("ts");
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                if ("home".equals(pageId) || "good_detail".equals(pageId)) {
                    if (mid != null && ts != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });
    }
}
